package cn._51doit

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * ┏┓　　　┏┓
 * ┏┛┻━━━┛┻┓
 * ┃　　　　　　　┃ 　
 * ┃　　　━　　　┃
 * ┃　┳┛　┗┳　┃
 * ┃　　　　　　　┃
 * ┃　　　┻　　　┃
 * ┃　　　　　　　┃
 * ┗━┓　　　┏━┛
 *    　　┃　　　┃神兽保佑
 *    　　┃　　　┃代码无BUG！
 *    　　┃　　　┗━━━┓
 *    　　┃　　　　　　　┣┓
 *    　　┃　　　　　　　┏┛
 *    　　┗┓┓┏━┳┓┏┛
 *    　　　┃┫┫　┃┫┫
 *    　　　┗┻┛　┗┻┛
 *
 * @create: 2021-10-11 22:38
 * @author: 今晚打脑斧
 * @program: kafka
 * @Description:
 *      用来测试生产者写入kafka的方式
 *    1.到底是调用一次send方法就理解向Kafka发送数据还是批量写入？
 *      答：没有立即写入，会先在客户端的buffer中缓存，当达到一定的大小或调用flush、close才会将数据写入到Kafka中
 *    2.如果不指定分区编号和key，那么到底是一直往一个分区中写入函数会切换分区？是写一条切换一次分区，还是达到一定数量再切换？
 *      答：默认是有的策略是轮询，当数据达到指定的大小、或调用flush，就会将数据写入到分区内，然后切换分区（目的是为了让数据均由分散到多个分区，就为了负载均衡）
 **/
object ScalaProducer2 {
  def main(args: Array[String]): Unit = {
    //指定broker的地址
    val properties = new Properties
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "doit01:9092,doit02:9092,doit03:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //创建一个生产者
    val producer = new KafkaProducer[String, String](properties)

    val producerRecord1 = new ProducerRecord[String, String]("wordcount", "hello kafka 1")
    producer.send(producerRecord1)

    val producerRecord2 = new ProducerRecord[String, String]("wordcount", "hello kafka 2")
    producer.send(producerRecord2)

    val producerRecord3 = new ProducerRecord[String, String]("wordcount", "hello kafka 3")
    producer.send(producerRecord3)

    producer.flush()

    val producerRecord4 = new ProducerRecord[String, String]("wordcount", "hello kafka 3")
    producer.send(producerRecord4)

    producer.close()
  }
}
